package com.zhu.utils;

import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.zhu.config.JDBCConfig;
import org.apache.commons.lang.StringUtils;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Collection;
import java.util.Set;

/**
 * PhoenixUtils
 * Phoenix 新增和修改 都为 upsert，底层基于hbase,新增和修改都为put操作
 * 处理工具类异常  选择抛出
 */
public class PhoenixUtil {


    /**
     * 插入数据
     * @param connection  数据库连接
     * @param sinkTable   Phoenix表名
     * @param data        表中数据   {"id":1001,"name":"zhu","sex":"man" }
     */
    public static void upsertValue(DruidPooledConnection connection, String sinkTable, JSONObject data)  {
        //sql upsert into db.tn(id,name,sex) value(' 1','zhu','man ')
        Set<String> columns = data.keySet();
        Collection<Object> values = data.values();
        //scala mkString 指定分隔符
        /*
        {"sinkTable":"dim_base_trademark","database":"flink","xid":3941,"data":{"tm_name":"中国李宁","id":14},
        "commit":true,"type":"insert","table":"base_trademark","ts":1679741043}
         */
        /*
        String upsertSql = "upsert into " + JDBCConfig.HBASE_SCHEMA + "." + sinkTable
                + "(" +
                StringUtils.join(columns,",")
                + ")"
                + " values ("
                + "'"
                + StringUtils.join(values,"','")
                + "')";

         */
        // 获取字段名
// 获取字段对应的值

// 拼接字段名
        String columnStr = StringUtils.join(columns, ",");
// 拼接字段值
        String valueStr = StringUtils.join(values, "','");
        System.out.println(columnStr);
        System.out.println(valueStr);
// 拼接插入语句
        String sql = "upsert into " + JDBCConfig.HBASE_SCHEMA
                + "." + sinkTable + "(" +
                columnStr + ") values ('" + valueStr + "')";
        //预编译
        PreparedStatement preparedStatement = null;
        try {
            preparedStatement = connection.prepareStatement(sql);
            preparedStatement.execute();
            connection.commit();  //提交
        }catch (SQLException sqlException){
            sqlException.printStackTrace();
            throw new RuntimeException("写入失败");
        }finally {
            if(preparedStatement !=null){
                try {
                    preparedStatement.close();
                }catch (SQLException sqlException){
                    sqlException.printStackTrace();
                }
            }
        }

        //执行
        //释放资源
    }
}
